package com.atguigu.edu.realtime220815.app.dwd.db;

import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/18 10:55
 **/
public class DwdExamQuestionDetail {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        tableEnv.executeSql(KafkaUtils.getTopicDbDDL("dwd_exam_question_detail"));

        //读取测验问题表
        Table test_exam_question = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['exam_id'] exam_id,\n" +
                "data['paper_id'] paper_id,\n" +
                "data['question_id'] question_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['is_correct'] is_correct,\n" +
                "data['score'] score,\n" +
                "date_format(data['create_time'], 'yyyy-MM-dd') date_id,\n" +
                "ts\n" +
                "from `topic_db` where `table` = 'test_exam_question' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("test_exam_question",test_exam_question);

        tableEnv.executeSql("CREATE TABLE dwd_exam_question_detail (\n" +
                "    id string,\n" +
                "    exam_id string,\n" +
                "    paper_id string,\n" +
                "    question_id string,\n" +
                "    user_id string,\n" +
                "    is_correct string,\n" +
                "    score string,\n" +
                "    date_id string,\n" +
                "    ts string,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + KafkaUtils.getUpsertKafkaDDL("dwd_exam_question_detail"));

        tableEnv.executeSql("insert into dwd_exam_question_detail select * from test_exam_question");


    }
}
